/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore.index;


import java.io.IOException;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;

import org.apache.cassandra.db.marshal.IntegerType;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.thrift.TException;
import cn.edu.thu.laud.objectstore.index.CompositeUtil.CompareType;

/**
 * 
 * @author zhuoan
 *
 */
public class IndexHandler {

	public static final byte VALUE_CODE_BYTES = 0;
	public static final byte VALUE_CODE_UTF8 = 1;
	public static final byte VALUE_CODE_UUID = 2;
	public static final byte VALUE_CODE_INT = 3;
	public static final byte VALUE_CODE_MAX = 127;

	public static final int DEFAULT_COUNT = 10000;
	public static final int ALL_COUNT = 100000;

	private static long lastTime = -1L;

	public static final IndexMeta defaultMeta = new IndexMeta();
	
	public static UUID newTimeUUID() {
		com.eaio.uuid.UUID eaioUUID = new com.eaio.uuid.UUID();
		return new UUID(eaioUUID.time, eaioUUID.clockSeqAndNode);
	}
	
	/**
	 * The Cassandra DynamicCompositeType will complain if component values of
	 * two different types are attempted to be compared. The way to prevent this
	 * and still allow for indexes to store different dynamic values is have a
	 * value code component that precedes the actual indexed value component in
	 * the composite. The DynamicCompositeType will first compare the two
	 * components holding the value codes, and if they don't match, then won't
	 * compare the next pair of components, avoiding the DynamicCompositeType
	 * throwing an error.
	 * 
	 * @param value
	 * @return value code
	 */
	public static byte getIndexableValueCode(Object value) {
		if (value instanceof String) {
			return VALUE_CODE_UTF8;
		} else if (value instanceof UUID) {
			return VALUE_CODE_UUID;
		} else if (value instanceof Number) {
			return VALUE_CODE_INT;
		} else {
			return VALUE_CODE_BYTES;
		}
	}
	
	public static String getIndexableValueString(Object value) {
		if (value instanceof ByteBuffer) {
			return ByteBufferUtil.bytesToHex((ByteBuffer)value);
		}
        return value.toString();
	}
	
	public static Object getIndexableValue(String str, int code) {
		if (code == VALUE_CODE_BYTES) {
			return ByteBufferUtil.hexToBytes(str);
		}else if (code == VALUE_CODE_UUID){
			return UUID.fromString(str);
		}else if (code == VALUE_CODE_INT){
			return Long.parseLong(str);
		}
		return str;
	}

	public static ByteBuffer getIndexableValueByteBuffer(Object value) {
		if (value instanceof String) {
			return ByteBufferUtil.bytes((String) value);
		} else if (value instanceof UUID) {
			return ByteBuffer.wrap(UUIDGen.decompose((UUID) value));
		} else if (value instanceof Number) {
			return IntegerType.instance.decompose(BigInteger
					.valueOf(((Number) value).longValue()));
		} else {
			return (ByteBuffer) value;
		}
	}
	


	private static void addIndexInsertion(BatchMutation batch,
			String columnIndexKey, ByteBuffer kvColumn, Object columnValue,
			UUID ts_uuid, long timestamp, IndexMeta cf) throws IOException {

		ByteBuffer compositeColumn = CompositeUtil
				.createDynamicCompositeKey(new Object[] {
						getIndexableValueCode(columnValue), columnValue,
						kvColumn, ts_uuid });

		batch.addInsertion(ByteBufferUtil.bytes(columnIndexKey), Arrays
				.asList(new String[] { cf.getIndex() }), new Column(
				compositeColumn).setValue(ByteBufferUtil.EMPTY_BYTE_BUFFER)
				.setTimestamp(timestamp));

	}


	private static void addIndexDeletion(BatchMutation batch,
			String columnIndexKey, ByteBuffer kvColumn, Object columnValue,
			UUID prev_timestamp, long timestamp, IndexMeta cf) throws IOException {

		ByteBuffer compositeColumn = CompositeUtil
				.createDynamicCompositeKey(new Object[] {
						getIndexableValueCode(columnValue), columnValue,
						kvColumn, prev_timestamp });
		
		SlicePredicate sp = new SlicePredicate();
		sp.addToColumn_names(compositeColumn);

		batch.addDeletion(ByteBufferUtil.bytes(columnIndexKey), Arrays
				.asList(new String[] { cf.getIndex() }), new Deletion()
				.setTimestamp(timestamp).setPredicate(sp));
	}

	private static void addEntriesInsertion(BatchMutation batch,
			ByteBuffer rowKey, Object columnName, Object columnValue,
			UUID ts_uuid, long timestamp, IndexMeta cf) throws IOException {

		batch.addInsertion(
				rowKey,
				Arrays.asList(new String[] { cf.getEntries() }),
				new Column(CompositeUtil
						.createDynamicCompositeKey(new Object[] { columnName,
								ts_uuid }))
						.setValue(
								CompositeUtil
										.createDynamicCompositeKey(new Object[] { columnValue }))
						.setTimestamp(timestamp));

	}

	private static void addEntriesDeletion(BatchMutation batch,
			ByteBuffer rowKey, ByteBuffer columnName, UUID prev_timestamp, long timestamp, IndexMeta cf) throws IOException {

		SlicePredicate sp = new SlicePredicate();
		sp.addToColumn_names(CompositeUtil
				.createDynamicCompositeKey(new Object[] { columnName,
						prev_timestamp }));

		batch.addDeletion(rowKey, Arrays
				.asList(new String[] { cf.getEntries() }), new Deletion()
				.setPredicate(sp).setTimestamp(timestamp));

	}
	
	/**
	 * update composite index for a column family
	 * 
	 */
	public static void updateCompositeIndex(Cassandra.Client client, BatchMutation batch, 
			KVColumn kvColumn, Object columnName, Object columnValue,
			IndexContainer container, IndexMeta cf, ConsistencyLevel wl)
			throws IOException, InvalidRequestException, UnavailableException,
			TimedOutException, TException {

		long timestamp = createClock();
		UUID ts_uuid = newTimeUUID();

		String columnIndexKey = container.getKey() + ":"
				+ columnName.toString();
		
		// Get all know previous index entries for this item's
		// indexed column from the item's index entry list
		ColumnParent parent = new ColumnParent(cf.getEntries());

		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange()
				.setStart(
						CompositeUtil.createDynamicCompositeKey(new Object[] {
								columnName, new UUID(0x0000000000001000L, 0L) }))
				.setFinish(
						CompositeUtil.createDynamicCompositeKey(new Object[] {
								columnName,
								new UUID(0xFFFFFFFFFFFF1FFFL, Long.MAX_VALUE
										| Long.MIN_VALUE) }))
				.setReversed(false).setCount(ALL_COUNT);
		predicate.setSlice_range(sliceRange);

		List<ColumnOrSuperColumn> columns = client.get_slice(
				kvColumn.getRowKey(), parent, predicate, wl);
		
		for (ColumnOrSuperColumn column : columns) {
			List<Object> deValue = CompositeUtil
					.deserialize(column.getColumn().name);
			UUID prev_timestamp = (UUID) deValue.get(1);
			// Delete all previous index entities from the item's index entry list
			addEntriesDeletion(batch, kvColumn.getRowKey(), column.getColumn().name, prev_timestamp, timestamp, cf);
			// Delete all previous index entities from both the container's
			// index
			Object prev_value = CompositeUtil.deserialize(
					column.getColumn().value).get(0);
			addIndexDeletion(batch, columnIndexKey, kvColumn.getBytes(),
					prev_value, prev_timestamp, timestamp, cf);
		}

		if (columnValue != null) {
			// Add the new index entry to the item's index entry list
			addEntriesInsertion(batch, kvColumn.getRowKey(), columnName, columnValue,
					ts_uuid, timestamp, cf);
			// Add the new index entry into the container's index
			addIndexInsertion(batch, columnIndexKey, kvColumn.getBytes(),
					columnValue, ts_uuid, timestamp, cf);
		}

		//client.batch_mutate(batch.getMutationMap(), wl);

	}
	
	/**
	 * build composite index for a column family
	 * 
	 */
	public static void buildCompositeIndex(Cassandra.Client client,
			ByteBuffer kvColumn, Object columnName, Object columnValue,
			IndexContainer container, IndexMeta cf, ConsistencyLevel wl)
			throws IOException, InvalidRequestException, UnavailableException,
			TimedOutException, TException {

		long timestamp = createClock();

		BatchMutation batch = new BatchMutation();
		UUID ts_uuid = newTimeUUID();

		// Add the new index entry to the item's index entry list		
		String columnIndexKey = container.getKey() + ":"
				+ columnName.toString();

		// Add the new index entry into the container's index

		if (columnValue != null) {
			addIndexInsertion(batch, columnIndexKey, kvColumn,
					columnValue, ts_uuid, timestamp, cf);
		}

		client.batch_mutate(batch.getMutationMap(), wl);

	}

	/**
	 * update composite index for a column family
	 * 
	 */
	public static void buildCompositeIndex(Cassandra.Client client,
			ByteBuffer kvColumn, Object columnName, Object columnValue,
			Set<IndexContainer> containers, IndexMeta cf, ConsistencyLevel wl)
			throws IOException, InvalidRequestException, UnavailableException,
			TimedOutException, TException {

		long timestamp = createClock();

		BatchMutation batch = new BatchMutation();
		UUID ts_uuid = newTimeUUID();

		// Add the new index entry to the item's index entry list
		for (IndexContainer container : containers) {

			String columnIndexKey = container.getKey() + ":"
					+ columnName.toString();

			// Add the new index entry into the container's index

			if (columnValue != null) {
				addIndexInsertion(batch, columnIndexKey, kvColumn,
						columnValue, ts_uuid, timestamp, cf);
			}

		}

		client.batch_mutate(batch.getMutationMap(), wl);

	}
	
	/**
	 * build composite index for a column family
	 * 
	 */
	public static void buildCompositeIndex(Cassandra.Client client,
			KVColumn kvColumn, Object columnName, IndexContainer container,
			IndexMeta cf, ConsistencyLevel wl) throws IOException,
			InvalidRequestException, UnavailableException, TimedOutException,
			TException {

		long timestamp = createClock();

		BatchMutation batch = new BatchMutation();
		UUID ts_uuid = newTimeUUID();
		//KVColumn kv = KVColumn.createKVColumn(kvColumn);

		// Get all know previous index entries for this item's
		// indexed column from the item's index entry list

		ColumnParent parent = new ColumnParent(cf.getEntries());

		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange()
				.setStart(
						CompositeUtil
								.createDynamicCompositeKey(new Object[] {
										columnName,
										new UUID(0x0000000000001000L, 0L) }))
				.setFinish(
						CompositeUtil.createDynamicCompositeKey(new Object[] {
								columnName,
								new UUID(0xFFFFFFFFFFFF1FFFL, Long.MAX_VALUE
										| Long.MIN_VALUE) }))
				.setReversed(false).setCount(ALL_COUNT);
		predicate.setSlice_range(sliceRange);

		List<ColumnOrSuperColumn> columns = client.get_slice(
				kvColumn.getRowKey(), parent, predicate, wl);

		// Delete all previous index entities from the item's index entry list

		for (ColumnOrSuperColumn column : columns) {
			List<Object> deValue = CompositeUtil
					.deserialize(column.getColumn().name);
			UUID prev_timestamp = (UUID) deValue.get(1);
			addEntriesDeletion(batch, kvColumn.getRowKey(), column.getColumn().name, prev_timestamp, timestamp, cf);
		}

		// Add the new index entry to the item's index entry list
		predicate = new SlicePredicate();
		predicate.setColumn_names(Arrays
						.asList(new ByteBuffer[] { getIndexableValueByteBuffer(columnName) }));
		parent = new ColumnParent(cf.getItem());

		List<ColumnOrSuperColumn> columnValueList = client.get_slice(
				kvColumn.getRowKey(), parent, predicate, wl);
		ByteBuffer columnValue = null;
		if (columnValueList != null && columnValueList.size() > 0) {
			columnValue = columnValueList.get(0).getColumn().value;
			addEntriesInsertion(batch, kvColumn.getRowKey(), columnName, columnValue,
					ts_uuid, timestamp, cf);
		}
		// else{
		// System.out.println("Error item key with no value found:"+itemKey);
		// }

		String columnIndexKey = container.getKey() + ":"
				+ columnName.toString();

		// Delete all previous index entities from both the container's
		// index

		for (ColumnOrSuperColumn column : columns) {

			List<Object> deValue = CompositeUtil.deserialize(column
					.getColumn().name);
			UUID prev_timestamp = (UUID) deValue.get(1);
			Object prev_value = CompositeUtil.deserialize(
					column.getColumn().value).get(0);

			addIndexDeletion(batch, columnIndexKey, kvColumn.getBytes(),
					prev_value, prev_timestamp, timestamp, cf);
		}

		// Add the new index entry into the container's index

		if (columnValue != null) {
			addIndexInsertion(batch, columnIndexKey, kvColumn.getBytes(),
					columnValue, ts_uuid, timestamp, cf);
		}

		
		client.batch_mutate(batch.getMutationMap(), wl);

	}

	/**
	 * build composite index for a column family
	 * 
	 */
	public static void buildCompositeIndex(Cassandra.Client client,
			KVColumn kvColumn, Object columnName, Set<IndexContainer> containers,
			IndexMeta cf, ConsistencyLevel wl) throws IOException,
			InvalidRequestException, UnavailableException, TimedOutException,
			TException {

		long timestamp = createClock();

		BatchMutation batch = new BatchMutation();
		UUID ts_uuid = newTimeUUID();
		//KVColumn kv = KVColumn.createKVColumn(kvColumn);

		// Get all know previous index entries for this item's
		// indexed column from the item's index entry list

		ColumnParent parent = new ColumnParent(cf.getEntries());

		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange()
				.setStart(
						CompositeUtil
								.createDynamicCompositeKey(new Object[] {
										columnName,
										new UUID(0x0000000000001000L, 0L) }))
				.setFinish(
						CompositeUtil.createDynamicCompositeKey(new Object[] {
								columnName,
								new UUID(0xFFFFFFFFFFFF1FFFL, Long.MAX_VALUE
										| Long.MIN_VALUE) }))
				.setReversed(false).setCount(ALL_COUNT);
		predicate.setSlice_range(sliceRange);

		List<ColumnOrSuperColumn> columns = client.get_slice(
				kvColumn.getRowKey(), parent, predicate, wl);

		// Delete all previous index entities from the item's index entry list

		for (ColumnOrSuperColumn column : columns) {
			List<Object> deValue = CompositeUtil
					.deserialize(column.getColumn().name);
			UUID prev_timestamp = (UUID) deValue.get(1);
			addEntriesDeletion(batch, kvColumn.getRowKey(), column.getColumn().name, prev_timestamp, timestamp, cf);
		}

		// Add the new index entry to the item's index entry list
		predicate = new SlicePredicate();
		predicate.setColumn_names(Arrays
						.asList(new ByteBuffer[] { getIndexableValueByteBuffer(columnName) }));
		parent = new ColumnParent(cf.getItem());

		List<ColumnOrSuperColumn> columnValueList = client.get_slice(
				kvColumn.getRowKey(), parent, predicate, wl);
		ByteBuffer columnValue = null;
		if (columnValueList != null && columnValueList.size() > 0) {
			columnValue = columnValueList.get(0).getColumn().value;
			addEntriesInsertion(batch, kvColumn.getRowKey(), columnName, columnValue,
					ts_uuid, timestamp, cf);
		}
		// else{
		// System.out.println("Error item key with no value found:"+itemKey);
		// }

		for (IndexContainer container : containers) {

			String columnIndexKey = container.getKey() + ":"
					+ columnName.toString();

			// Delete all previous index entities from both the container's
			// index

			for (ColumnOrSuperColumn column : columns) {

				List<Object> deValue = CompositeUtil.deserialize(column
						.getColumn().name);
				UUID prev_timestamp = (UUID) deValue.get(1);
				Object prev_value = CompositeUtil.deserialize(
						column.getColumn().value).get(0);

				addIndexDeletion(batch, columnIndexKey, kvColumn.getBytes(),
						prev_value, prev_timestamp, timestamp, cf);
			}

			// Add the new index entry into the container's index

			if (columnValue != null) {
				addIndexInsertion(batch, columnIndexKey, kvColumn.getBytes(),
						columnValue, ts_uuid, timestamp, cf);
			}

		}

		client.batch_mutate(batch.getMutationMap(), wl);

	}

	
	/**
	 * Sets the item column value for an item contained in a set of collections.
	 * 
	 * @throws TException
	 * @throws TimedOutException
	 * @throws UnavailableException
	 * @throws InvalidRequestException
	 * @throws IOException
	 * 
	 */
	public static void setItemColumn(Cassandra.Client client, KVColumn kvColumn,
			Object columnName, Object columnValue,
			IndexContainer container, IndexMeta cf, ConsistencyLevel wl)
			throws IOException, InvalidRequestException, UnavailableException,
			TimedOutException, TException {

		long timestamp = createClock();

		BatchMutation batch = new BatchMutation();
		UUID ts_uuid = newTimeUUID();

		// Get all know previous index entries for this item's
		// indexed column from the item's index entry list

		ColumnParent parent = new ColumnParent(cf.getEntries());

		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange()
				.setStart(
						CompositeUtil.createDynamicCompositeKey(new Object[] {
								columnName, new UUID(0x0000000000001000L, 0L) }))
				.setFinish(
						CompositeUtil.createDynamicCompositeKey(new Object[] {
								columnName,
								new UUID(0xFFFFFFFFFFFF1FFFL, Long.MAX_VALUE
										| Long.MIN_VALUE) }))
				.setReversed(false).setCount(ALL_COUNT);
		predicate.setSlice_range(sliceRange);

		List<ColumnOrSuperColumn> columns = client.get_slice(
				kvColumn.getRowKey(), parent, predicate, wl);

		// Delete all previous index entities from the item's index entry list

		for (ColumnOrSuperColumn column : columns) {
			List<Object> deValue = CompositeUtil
					.deserialize(column.getColumn().name);
			UUID prev_timestamp = (UUID) deValue.get(1);
			addEntriesDeletion(batch, kvColumn.getRowKey(), column.getColumn().name, prev_timestamp, timestamp, cf);
		}

		// Add the new index entry to the item's index entry list

		if (columnValue != null) {
			addEntriesInsertion(batch, kvColumn.getRowKey(), columnName, columnValue,
					ts_uuid, timestamp, cf);
		}


		String columnIndexKey = container.getKey() + ":"
				+ columnName.toString();

		// Delete all previous index entities from both the container's
		// index

		for (ColumnOrSuperColumn column : columns) {

			List<Object> deValue = CompositeUtil.deserialize(column
					.getColumn().name);
			UUID prev_timestamp = (UUID) deValue.get(1);
			Object prev_value = CompositeUtil.deserialize(
					column.getColumn().value).get(0);

			addIndexDeletion(batch, columnIndexKey, kvColumn.getBytes(),
					prev_value, prev_timestamp, timestamp, cf);
		}

		// Add the new index entry into the container's index

		if (columnValue != null) {
			addIndexInsertion(batch, columnIndexKey, kvColumn.getBytes(),
					columnValue, ts_uuid, timestamp, cf);
		}

		// Store the new column value into the item
		// If new value is null, delete the value instead

		if (columnValue != null) {

			batch.addInsertion(kvColumn.getRowKey(), Arrays
					.asList(new String[] { cf.getItem() }),
					new Column(getIndexableValueByteBuffer(columnName))
							.setValue(getIndexableValueByteBuffer(columnValue))
							.setTimestamp(timestamp));

		} else {

			SlicePredicate sp = new SlicePredicate();
			sp.addToColumn_names(getIndexableValueByteBuffer(columnName));

			batch.addDeletion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cf.getItem() }),
					new Deletion().setPredicate(sp).setTimestamp(timestamp));
		}

		client.batch_mutate(batch.getMutationMap(), wl);

	}
	
	/**
	 * Sets the item column value for an item contained in a set of collections.
	 * 
	 * @throws TException
	 * @throws TimedOutException
	 * @throws UnavailableException
	 * @throws InvalidRequestException
	 * @throws IOException
	 * 
	 */
	public static void setItemColumn(Cassandra.Client client, KVColumn kvColumn,
			Object columnName, Object columnValue,
			Set<IndexContainer> containers, IndexMeta cf, ConsistencyLevel wl)
			throws IOException, InvalidRequestException, UnavailableException,
			TimedOutException, TException {

		long timestamp = createClock();

		BatchMutation batch = new BatchMutation();
		UUID ts_uuid = newTimeUUID();

		// Get all know previous index entries for this item's
		// indexed column from the item's index entry list

		ColumnParent parent = new ColumnParent(cf.getEntries());

		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange()
				.setStart(
						CompositeUtil.createDynamicCompositeKey(new Object[] {
								columnName, new UUID(0x0000000000001000L, 0L) }))
				.setFinish(
						CompositeUtil.createDynamicCompositeKey(new Object[] {
								columnName,
								new UUID(0xFFFFFFFFFFFF1FFFL, Long.MAX_VALUE
										| Long.MIN_VALUE) }))
				.setReversed(false).setCount(ALL_COUNT);
		predicate.setSlice_range(sliceRange);

		List<ColumnOrSuperColumn> columns = client.get_slice(
				kvColumn.getRowKey(), parent, predicate, wl);

		// Delete all previous index entities from the item's index entry list

		for (ColumnOrSuperColumn column : columns) {
			List<Object> deValue = CompositeUtil
					.deserialize(column.getColumn().name);
			UUID prev_timestamp = (UUID) deValue.get(1);
			addEntriesDeletion(batch, kvColumn.getRowKey(), column.getColumn().name, prev_timestamp, timestamp, cf);
		}

		// Add the new index entry to the item's index entry list

		if (columnValue != null) {
			addEntriesInsertion(batch, kvColumn.getRowKey(), columnName, columnValue,
					ts_uuid, timestamp, cf);
		}

		for (IndexContainer container : containers) {

			String columnIndexKey = container.getKey() + ":"
					+ columnName.toString();

			// Delete all previous index entities from both the container's
			// index

			for (ColumnOrSuperColumn column : columns) {

				List<Object> deValue = CompositeUtil.deserialize(column
						.getColumn().name);
				UUID prev_timestamp = (UUID) deValue.get(1);
				Object prev_value = CompositeUtil.deserialize(
						column.getColumn().value).get(0);

				addIndexDeletion(batch, columnIndexKey, kvColumn.getBytes(),
						prev_value, prev_timestamp, timestamp, cf);
			}

			// Add the new index entry into the container's index

			if (columnValue != null) {
				addIndexInsertion(batch, columnIndexKey, kvColumn.getBytes(),
						columnValue, ts_uuid, timestamp, cf);
			}

		}

		// Store the new column value into the item
		// If new value is null, delete the value instead

		if (columnValue != null) {

			batch.addInsertion(kvColumn.getRowKey(), Arrays
					.asList(new String[] { cf.getItem() }),
					new Column(getIndexableValueByteBuffer(columnName))
							.setValue(getIndexableValueByteBuffer(columnValue))
							.setTimestamp(timestamp));

		} else {

			SlicePredicate sp = new SlicePredicate();
			sp.addToColumn_names(getIndexableValueByteBuffer(columnName));

			batch.addDeletion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cf.getItem() }),
					new Deletion().setPredicate(sp).setTimestamp(timestamp));
		}

		client.batch_mutate(batch.getMutationMap(), wl);

	}

	/**
	 * Search container.
	 * 
	 * @throws IOException
	 * @throws TException 
	 * @throws TimedOutException 
	 * @throws UnavailableException 
	 * @throws InvalidRequestException 
	 * 
	 */
	public static Set<KVColumn> searchContainer(Cassandra.Client client, Object columnName, Object searchValue,
			KVColumn startResult, int count, boolean reversed,
			IndexContainer container, IndexMeta cf,
			ConsistencyLevel rl) throws IOException, InvalidRequestException, UnavailableException, TimedOutException, TException {

		return searchContainer(client, columnName, searchValue, true,
				searchValue, true, startResult, count, reversed, false, null, null, null, container, cf, rl);
	}

	public static long createClock() {

		long timestamp = System.currentTimeMillis();
		synchronized (IndexHandler.class) {
			if (timestamp > lastTime) {
				lastTime = timestamp;
			} else {
				timestamp = ++lastTime;
			}
		}

		return timestamp;
	}
	
	/**
	 * Search container.
	 * 
	 * @throws IOException
	 * @throws TException
	 * @throws TimedOutException
	 * @throws UnavailableException
	 * @throws InvalidRequestException
	 * 
	 */
	public static Set<KVColumn> searchContainer(Cassandra.Client client, Object columnName, Object startValue,
			boolean incStart, Object endValue, boolean incEnd, KVColumn startResult, int count,
			boolean reversed, boolean page, List<Object> valueList, ByteBuffer disColName, Set<ByteBuffer> disColValues,
			IndexContainer container, IndexMeta cf, ConsistencyLevel rl)
			throws IOException, InvalidRequestException, UnavailableException, TimedOutException, TException {

		Set<KVColumn> items = new LinkedHashSet<KVColumn>();

		String columnIndexKey = container.getKey() + ":"
				+ columnName.toString();

		if (count == 0) {
			count = DEFAULT_COUNT;
		}

		ByteBuffer start = null;
		if (startValue == null) {
			if (startResult != null) {
				start = CompositeUtil.createDynamicCompositeKey(new Object[] {
						reversed ? VALUE_CODE_MAX : VALUE_CODE_BYTES, ByteBufferUtil.EMPTY_BYTE_BUFFER, startResult.getBytes() });
			} else {
				start = CompositeUtil.createDynamicCompositeKey(new Object[] {
						reversed ? VALUE_CODE_MAX : VALUE_CODE_BYTES, ByteBufferUtil.EMPTY_BYTE_BUFFER });
			}
		} else if (startResult != null) {
			if(incStart)
				start = CompositeUtil
						.createDynamicCompositeKey(new Object[] {
								getIndexableValueCode(startValue), startValue,
								startResult.getBytes()}, new CompareType[]{CompareType.EQUAL, CompareType.EQUAL, reversed ? CompareType.GREATER_THAN_EQUAL : CompareType.EQUAL });
			else
				start = CompositeUtil
				.createDynamicCompositeKey(new Object[] {
						getIndexableValueCode(startValue), startValue,
						startResult.getBytes()}, new CompareType[]{ CompareType.EQUAL, reversed ? CompareType.LESS_THAN_EQUAL: CompareType.GREATER_THAN_EQUAL, CompareType.EQUAL });
		} else {
			if(incStart)
				start = CompositeUtil.createDynamicCompositeKey(new Object[] {
						getIndexableValueCode(startValue), startValue }, new CompareType[] {CompareType.EQUAL, reversed ? CompareType.GREATER_THAN_EQUAL : CompareType.EQUAL});
			else
				start = CompositeUtil.createDynamicCompositeKey(new Object[] {
						getIndexableValueCode(startValue), startValue }, new CompareType[] {CompareType.EQUAL, reversed ? CompareType.LESS_THAN_EQUAL: CompareType.GREATER_THAN_EQUAL});
		}

		ByteBuffer finish = null;

		if (endValue != null) {
			if (incEnd)
				finish = CompositeUtil.createDynamicCompositeKey(new Object[] {
						getIndexableValueCode(endValue), endValue },
						new CompareType[] { CompareType.EQUAL,
								reversed ? CompareType.EQUAL : CompareType.GREATER_THAN_EQUAL });
			
			else
				finish = CompositeUtil.createDynamicCompositeKey(new Object[] {
						getIndexableValueCode(endValue), endValue }, new CompareType[] { CompareType.EQUAL,
								reversed ? CompareType.GREATER_THAN_EQUAL : CompareType.EQUAL });
		} else{
			finish = CompositeUtil.createDynamicCompositeKey(new Object[] {
					reversed ? VALUE_CODE_BYTES : VALUE_CODE_MAX, ByteBufferUtil.EMPTY_BYTE_BUFFER});
		}

		ColumnParent parent = new ColumnParent(cf.getIndex());

		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange().setStart(start)
				.setFinish(finish).setReversed(reversed).setCount(count);

		predicate.setSlice_range(sliceRange);
		List<ColumnOrSuperColumn> columns = null;
	
		columns = client.get_slice(ByteBufferUtil.bytes(columnIndexKey),
				parent, predicate, rl);
		
        Object lastValue = null;
        ByteBuffer lastResult = null;
        boolean more = count>1 && columns!=null && columns.size()>=count;
        
		if (columns != null &&columns.size()>0) {
			for (int i =0; i<columns.size()-1; i++) {
				ColumnOrSuperColumn result = columns.get(i); 
				List<Object> deValue = CompositeUtil.deserialize(result
						.getColumn().name);
				//Object value = deValue.get(1);
				ByteBuffer kvBytes = (ByteBuffer) deValue.get(2);
				if (kvBytes != null) {
					KVColumn kv = KVColumn.createKVColumn(kvBytes);
					if(disColValues != null && !disColValues.contains(kv.getColumnValue(disColName)))
						continue;
					if(items.add(kv) && valueList != null)
						valueList.add(deValue.get(1));
					
				}
			}
			List<Object> lastCol = CompositeUtil.deserialize(columns.get(columns.size()-1).getColumn().name);
			if(more && page){				
				lastResult = (ByteBuffer)lastCol.get(2);
				lastValue = lastCol.get(1);
			}else {
				KVColumn kv = KVColumn.createKVColumn((ByteBuffer)lastCol.get(2));
				if(disColValues == null || disColValues.contains(kv.getColumnValue(disColName))){
					if(items.add(kv) && valueList != null)
						valueList.add(lastCol.get(1));
				}
			}
			
		}
		
		if(page){
			
			String externKey = (lastValue == null||!more ? "null":(char)(getIndexableValueCode(lastValue))+getIndexableValueString(lastValue))+ ":"+(lastResult == null||!more ? "null":getIndexableValueString(lastResult))+":"+Boolean.toString(more);
			cf.setPageMeta(externKey);
		}

		return items;
	}

	/**
	 * Adds the item to collection.
	 * 
	 * @throws TException
	 * @throws TimedOutException
	 * @throws UnavailableException
	 * @throws InvalidRequestException
	 * 
	 */
	public static void addItemToContainer(Cassandra.Client client, String itemKey,
			IndexContainer container, IndexMeta cf,
			ConsistencyLevel wl) throws InvalidRequestException,
			UnavailableException, TimedOutException, TException {

		BatchMutation batch = new BatchMutation();
		batch.addInsertion(ByteBufferUtil.bytes(container.getKey()), Arrays
				.asList(new String[] { cf.getItems() }), new Column(
				ByteBufferUtil.bytes(itemKey))
				.setValue(getIndexableValueByteBuffer(createClock())));
		client.batch_mutate(batch.getMutationMap(), wl);

	}

	public static List<String> getItemsInContainer(Cassandra.Client client,
			IndexContainer container, IndexMeta cf, ConsistencyLevel rl) throws InvalidRequestException, UnavailableException, TimedOutException, TException {
		List<String> keys = new ArrayList<String>();

		ColumnParent parent = new ColumnParent(cf.getItems());

		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange().setStart(ByteBufferUtil.EMPTY_BYTE_BUFFER)
				.setFinish(ByteBufferUtil.EMPTY_BYTE_BUFFER).setReversed(false)
				.setCount(ALL_COUNT);

		predicate.setSlice_range(sliceRange);

		List<ColumnOrSuperColumn> columns = null;
		
		columns = client.get_slice(
				ByteBufferUtil.bytes(container.getKey()), parent,
				predicate, rl);
		
		if (columns != null) {
			for (ColumnOrSuperColumn column : columns) {
				keys.add(new String(column.getColumn().getName()));
			}
		}
		return keys;
	}

}
